package shoppingAnalysis;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import java.io.IOException;
import java.net.URI;

public class Task3 {

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);// 设置环境参数
		job.setJarByClass(Task3.class);// 设置整个程序的类名

		job.setMapperClass(Scan_numsMapper.class);
		job.setReducerClass(Scan_numsReducer.class);
		job.setCombinerClass(MyCombiner3.class);
		job.setPartitionerClass(MyPartitioner3.class);

		// reduce 输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		// 设置自定义的格式化输出
		job.setOutputFormatClass(ShopFileOutputFormat.class);
		job.setNumReduceTasks(34);// 全国34个省份

		FileInputFormat.setInputPaths(job,
				new Path("hdfs://192.168.3.101:9000/ShoppingAnalysis/input/raw_user_table.txt"));
		// 如果文件系统已存在输出文件夹则删除
		FileSystem fs = FileSystem.get(new URI("hdfs://192.168.3.101:9000"), conf, "roger");
		if (fs.exists(new Path("/ShoppingAnalysis/test"))) {
			fs.delete(new Path("/ShoppingAnalysis/test"), true);
		}

		// 注意这里不是FileOutPutFormat，下面同样注意
		ShopFileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.3.101:9000/ShoppingAnalysis/test"));

		/*
		 * 消除系统的默认输出文件，不过因为文件的输出格式变了，不再是FileOutPutFormat，所以这里也要改成相对应的文件输出格式
		 */
		LazyOutputFormat.setOutputFormatClass(job, ShopFileOutputFormat.class);

		job.getConfiguration().setStrings("mapreduce.reduce.shuffle.memory.limit.percent", "0.1");

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}

class Scan_numsMapper extends Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		String[] words = value.toString().split("\\s+");
		// 统计浏览量
		if (words[3].equals("1")) {
			// key值：省份 +日期 。value: 数字 1
			context.write(new Text(words[6] + "\t" + words[5]), new Text("1"));

		}
	}
}

class Scan_numsReducer extends Reducer<Text, Text, Text, Text> {

	private MultipleOutputs<Text, Text> multipleOutputs;

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		multipleOutputs = new MultipleOutputs<Text, Text>(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

		String[] words = key.toString().split("\\s+");

		// 输出当前省份 所有日期对应的浏览量
		// 进一步reduce
		int sum = 0;
		for (Text value : values) {
			sum += Integer.parseInt(value.toString());
		}
		// 输出：key:省份。value:日期，浏览量
		multipleOutputs.write(new Text(words[0]), new Text(words[1] + "\t" + String.valueOf(sum)), words[0]);

	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub

		multipleOutputs.close();
	}
}
